/*
 * Copyright 2010 - Jeremie Brebec
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */
package test.org.jb.notification;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import org.jb.notification.client.Notification;
import org.jb.notification.client.Notifications;
import org.jb.notification.client.rpc.NotificationService;

public class ConsumersTester {

	public static interface Consumer {
		void handle( List<Notification> notifications );
	}

	
	private class ConsumerThread extends Thread {
		
		public ConsumerThread( Consumer consumer ) {
			this.consumer = consumer;
		}
		
		@Override
		public void run() {
			try {
				while( ! stop ) {
					
					Notifications notifications = notificationService.getNotification(checkpoint);
					
					checkpoint = notifications.getCheckpoint();
					
					if (notifications.getNotifications() != null && !notifications.getNotifications().isEmpty())
						consumer.handle(notifications.getNotifications());
				}
			}
			catch( Exception e ) {
				exception = e;
				throw new RuntimeException(e);
			}
		}
		
		public void stopConsumer() {
			stop = true;
		}
		
		public void rethrow() throws Exception {
			if (exception != null)
				throw exception;
		}

		private final Consumer consumer;
		private boolean stop = false;
		private long checkpoint = 0;
		private Exception exception;
	}
	
	public void addConsumer( Consumer consumer ) {
		threads.add( new ConsumerThread(consumer) );
	}
	
	public void setNotificationService( NotificationService service ) {
		this.notificationService = service;
	}
	
	public void run() {
		for( ConsumerThread thread : threads ) {
			thread.start();
		}
	}
	
	public void finish() throws Exception {
		for( ConsumerThread thread : threads ) {
			thread.stopConsumer();
		}
		
		for( ConsumerThread thread : threads ) {
			thread.join( 10 * 1000 );
		}
		
		for( ConsumerThread thread : threads ) {
			thread.rethrow();
		}
	}
	
	
	
	private NotificationService notificationService;
	private Collection<ConsumerThread> threads = new ArrayList<ConsumerThread>();
}
